/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

 

package com.sui.bigdata.flink.sql.core.sink;

import com.sui.bigdata.flink.sql.core.table.AbsTableParser;
import com.sui.bigdata.flink.sql.core.table.TargetTableInfo;
import com.sui.bigdata.flink.sql.sink.console.ConsoleSink;
import com.sui.bigdata.flink.sql.sink.console.table.ConsoleSinkParser;
import com.sui.bigdata.flink.sql.sink.hbase.HbaseSink;
import com.sui.bigdata.flink.sql.sink.kafka.KafkaSink;
import com.sui.bigdata.flink.sql.sink.kafka.table.KafkaSinkParser;
import com.sui.bigdata.flink.sql.sink.kudu.KuduSink;
import com.sui.bigdata.flink.sql.sink.kudu.table.KuduSinkParser;
import com.sui.bigdata.flink.sql.sink.message.MessageSink;
import com.sui.bigdata.flink.sql.sink.message.table.MessageSinkParser;
import com.sui.bigdata.flink.sql.sink.mysql.MysqlSink;
import com.sui.bigdata.flink.sql.sink.mysql.table.MysqlSinkParser;
import com.sui.bigdata.flink.sql.sink.phoenix.PhoenixSink;
import com.sui.bigdata.flink.sql.sink.phoenix.table.PhoenixSinkParser;
import com.sui.bigdata.flink.sql.sink.redis.RedisSink;
import com.sui.bigdata.flink.sql.sink.redis.table.RedisSinkParser;
import org.apache.flink.table.sinks.TableSink;
import com.sui.bigdata.flink.sql.sink.hbase.HbaseSinkParser;

/**
 * Loads jar and initializes the object according to the specified sink type
 * Date: 2019/07/10
 * @author yong_chen@sui.com
 */

public class StreamSinkFactory {

    private static final String HBASE = "hbase";
    private static final String CONSOLE = "console";
    private static final String REDIS = "redis";
    private static final String KAFKA = "kafka";
    private static final String MYSQL = "mysql";
    private static final String PHOENIX = "phoenix";
    private static final String KUDU = "kudu";
    private static final String MESSAGE = "message";


    public static AbsTableParser getSqlParser(String pluginType) throws Exception {

        switch (pluginType){
            case HBASE:
                return new HbaseSinkParser();
            case CONSOLE:
                return new ConsoleSinkParser();
            case REDIS:
                return new RedisSinkParser();
            case KAFKA:
                return new KafkaSinkParser();
            case MYSQL:
                return new MysqlSinkParser();
            case KUDU:
                return new KuduSinkParser();
            case MESSAGE:
                return new MessageSinkParser();
            case PHOENIX:
                return new PhoenixSinkParser();
            default:
                throw new RuntimeException(" not support this sink now ");
        }
    }

    public static TableSink getTableSink(TargetTableInfo targetTableInfo) throws Exception {
        IStreamSinkGener streamSinkGener;
        Object result;
        switch (targetTableInfo.getType()){
            case HBASE:
                streamSinkGener = new HbaseSink();
                result = streamSinkGener.genStreamSink(targetTableInfo);
                return (TableSink) result;
            case CONSOLE:
                streamSinkGener = new ConsoleSink();
                result = streamSinkGener.genStreamSink(targetTableInfo);
                return (TableSink) result;
            case REDIS:
                 streamSinkGener = new RedisSink();
                 result = streamSinkGener.genStreamSink(targetTableInfo);
                 return (TableSink) result;
            case KAFKA:
                 streamSinkGener = new KafkaSink();
                 result = streamSinkGener.genStreamSink(targetTableInfo);
                 return (TableSink) result;
            case MYSQL:
                 streamSinkGener = new MysqlSink();
                 result = streamSinkGener.genStreamSink(targetTableInfo);
                 return (TableSink) result;
            case KUDU:
                streamSinkGener = new KuduSink();
                result = streamSinkGener.genStreamSink(targetTableInfo);
                return (TableSink) result;
            case MESSAGE:
                streamSinkGener = new MessageSink();
                result = streamSinkGener.genStreamSink(targetTableInfo);
                return (TableSink) result;
            case PHOENIX:
                streamSinkGener = new PhoenixSink();
                result = streamSinkGener.genStreamSink(targetTableInfo);
                return (TableSink) result;
            default:
                throw new RuntimeException(" not support this sink now ");
        }
    }

}
